package com.it.table;

import com.it.operator.utils.SourceUtils;
import com.it.pojo.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author code1997
 */
public class SimpleTableDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(1);
        SingleOutputStreamOperator<Event> eventSource = SourceUtils.getEventSource(executionEnvironment)
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                    @Override
                    public long extractTimestamp(Event element, long recordTimestamp) {
                        return element.timestamp;
                    }
                }));
        //创建表执行环境
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(executionEnvironment);
        //将DataStream转换成Table
        Table table = streamTableEnvironment.fromDataStream(eventSource);
        //方式1：直接写SQL进行转换
        Table resultTable = streamTableEnvironment.sqlQuery("select user,url,`timestamp` from " + table);
        //打印表结构
        resultTable.printSchema();
        //打印表数据：result> +I[tom, ./order?id=10] +I代表insert。
        streamTableEnvironment.toDataStream(resultTable).print("result");

        Table result2 = table.select($("user"), $("url"))
                .where($("user").isEqual("tom"));
        //转换为流打印输出
        streamTableEnvironment.toDataStream(result2).print("result2");
        executionEnvironment.execute();
    }
}
